package com.yan.test;


import java.io.File;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created by yanl_ on 2018/3/10.
 *
 *
 * http://blog.csdn.net/u013668852/article/details/53264099
 */
public class Operation {


    private static LinkedBlockingDeque<File> files;//即将上传的文件队列
    private static ExecutorService UploadThreadPool= Executors.newFixedThreadPool(2);//上传文件线程池


    /**
     * 	@param path 需要上传的文件夹路径
     */
    public Operation(String path) {
        files=new LinkedBlockingDeque<File>();
        //读取文件列表
        ReadFiles(path);
    }

    /**
     * 读取文件列表到队列
     * @param path
     */
    private void ReadFiles(String path){
        System.out.println(path);
        File file=new File(path);
        File [] filen=file.listFiles();
        for(int i=0;i<filen.length;i++){
            files.add(filen[i]);
        }
    }


    /**
     * 将文件内容读取到队列中
     */
    public void MultiReadFiles(){
        while(files.size()>0){
            try {
                new Util(files.take());
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    }

    public void MultiUpload(){

        UploadThreadPool.execute(new Runnable() {

            public void run() {
                // TODO Auto-generated method stub
                while(Util.Contents.size()==0){
                    try {
                        System.out.println("等待一秒");
                        Thread.sleep(1000*1);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
                while(Util.Contents.size()>0){
                    System.out.println("now numbers:"+Util.Contents.size());
                    String content = null;
                    try{
                        content = Util.Contents.take();
                        //处理过程开始

                        //结束处理过程
                    }catch(Exception e){
                        try {
                            e.printStackTrace();
                            //插入失败，内容重回队列，准备再次插入
                            Util.Contents.put(content);
                        } catch (InterruptedException e1) {
                            // TODO Auto-generated catch block
                            e1.printStackTrace();
                        }
                    }
                    while(Util.Contents.size()==0&&files.size()>0){
                        try {
                            System.out.println("等待一秒");
                            Thread.sleep(1000*1);
                        } catch (InterruptedException e) {
                            // TODO Auto-generated catch block
                            e.printStackTrace();
                        }
                    }
                }
                System.out.println("now numbers:"+Util.Contents.size());
                try {
                    Thread.sleep(1000*2);//线程终止2秒
                    if(Util.Contents.size()==0&&files.size()==0){
                        System.exit(0);
                    }
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });
    }



    private static String path="D:/个人材料/datetest"; //日志预处理：需要扫描的源文件

    public static void main(String[] args)
    {
//		Transform transform = new Transform();
//		transform.CreateHbTable();							//在Hbase中创建表
        Operation operation = new Operation(path);
        //operation.insertTest();  //数据录入及预处理 ——日志数据导入到表中
        operation.MultiReadFiles();
        operation.MultiUpload();
        //ConsumeDML(operation);  	                        //按照消费等级对用户分类
    }

}